package com.fiona.remoting.netty;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fiona.core.message.Message;
import com.fiona.core.message.RpcRequest;
import com.fiona.core.message.RpcResponse;
import com.fiona.core.message.enums.SerializationTypeEnum;
import com.fiona.remoting.api.Channel;

import io.netty.channel.ChannelFutureListener;

public class NettyChannel implements Channel {

	private static final Logger logger = LoggerFactory.getLogger(NettyChannel.class);

	private static final ConcurrentMap<io.netty.channel.Channel, NettyChannel> CHANNEL_MAP = new ConcurrentHashMap<io.netty.channel.Channel, NettyChannel>();
	private static io.netty.channel.Channel channel = null;
	private final Map<String, Object> attributes = new ConcurrentHashMap<String, Object>();
	private final AtomicBoolean active = new AtomicBoolean(false);

	private NettyChannel(io.netty.channel.Channel channel) {
		if (channel == null) {
			throw new IllegalArgumentException("netty channel == null;");
		}
		this.channel = channel;
	}

	public static NettyChannel getOrAddChannel(io.netty.channel.Channel channel) {
		if (channel == null) {
			return null;
		}
		NettyChannel ret = CHANNEL_MAP.get(channel);
		if (ret == null) {
			NettyChannel nettyChannel = new NettyChannel(channel);
			if (channel.isActive()) {
				nettyChannel.markActive(true);
				ret = CHANNEL_MAP.putIfAbsent(channel, nettyChannel);
			}
			if (ret == null) {
				ret = nettyChannel;
			}
		}
		return ret;
	}

	public static void removeChannelIfDisconnected(io.netty.channel.Channel channel) {
		if (channel != null && !channel.isActive()) {
			NettyChannel nettyChannel = CHANNEL_MAP.remove(channel);
			if (nettyChannel != null) {
				nettyChannel.markActive(false);
			}
		}
	}

	static void removeChannel(io.netty.channel.Channel channel) {
		if (channel != null) {
			NettyChannel nettyChannel = CHANNEL_MAP.remove(channel);
			if (nettyChannel != null) {
				nettyChannel.markActive(false);
			}
		}
	}

	@Override
	public boolean isConnected() {
		return active.get();
	}

	public boolean isActive() {
		return active.get();
	}

	public void markActive(boolean isActive) {
		active.set(isActive);
	}

	@Override
	public InetSocketAddress remoteAddress() {
		return (InetSocketAddress) channel.remoteAddress();
	}

	@Override
	public InetSocketAddress localAddress() {
		return (InetSocketAddress) channel.localAddress();
	}

	@Override
	public void close() {
        try {
            removeChannelIfDisconnected(channel);
        } catch (Exception e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            attributes.clear();
        } catch (Exception e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            if (logger.isInfoEnabled()) {
                logger.info("Close netty channel " + channel);
            }
            channel.close();
        } catch (Exception e) {
            logger.warn(e.getMessage(), e);
        }
	}

	@Override
	public Object send(Object message) {
		RpcRequest rpcRequest = (RpcRequest) message;
		// build return value
        CompletableFuture<RpcResponse> resultFuture = new CompletableFuture<>();
        // build rpc service name by rpcRequest
        String rpcServiceName = rpcRequest.getInterfaceName();
        // get server address
        //InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcServiceName);
        // get  server address related channel
        if (channel.isActive()) {
            // put unprocessed request
           // unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);
            Message rpcMessage = new Message();
            rpcMessage.setData(rpcRequest);
            rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode());
            rpcMessage.setType(Constants.REQUEST_TYPE);
            channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                	logger.info("client send message: [{}]", rpcMessage);
                } else {
                    future.channel().close();
                    resultFuture.completeExceptionally(future.cause());
                    logger.error("Send failed:", future.cause());
                }
            });
        } else {
            throw new IllegalStateException();
        }

        return resultFuture;
    }
	

    public boolean hasAttribute(String key) {
        return attributes.containsKey(key);
    }

    public Object getAttribute(String key) {
        return attributes.get(key);
    }

    public void setAttribute(String key, Object value) {
        if (value == null) {
            attributes.remove(key);
        } else {
            attributes.put(key, value);
        }
    }

}
